//
// Created by benxb on 2021/11/16.
//

#include "storage/disk/data_chunk_desc.h"
#include "common/util/file_util.h"
#include "common/exception.h"
#include "common/logger.h"

namespace benxdb {

DataChunkDescIO::~DataChunkDescIO() {
    desc_io_.close();
}

DataChunkDescIO::DataChunkDescIO(const std::string& table_name) : table_name_(table_name) {
    desc_file_ = GlobalConfig::DescFilePath(table_name);
    desc_io_.open(desc_file_, std::ios::binary | std::ios::in | std::ios::out);
    // directory or file does not exist
    if (!desc_io_.is_open()) {
        desc_io_.clear();
        // create a new file
        desc_io_.open(desc_file_, std::ios::binary | std::ios::trunc | std::ios::out);
        desc_io_.close();
        // reopen with original mode
        desc_io_.open(desc_file_, std::ios::binary | std::ios::in | std::ios::out);
        if (!desc_io_.is_open()) {
            throw Exception("can't open desc file");
        }
    }
    file_size_ = FileUtil::GetFileSize(desc_file_);
}

uint64_t DataChunkDescIO::AppendDesc(ChunkDesc desc) {
    uint64_t chunk_id = GetChunkCount();
    desc_io_.seekp(0, std::ios_base::end);
    desc_io_.write(reinterpret_cast<const char *>(&desc.chunk_offset), sizeof(desc.chunk_offset));
    desc_io_.write(reinterpret_cast<const char *>(&desc.chunk_size), sizeof(desc.chunk_size));
    desc_io_.write(reinterpret_cast<const char *>(&desc.type), sizeof(desc.type));
    desc_io_.write(reinterpret_cast<const char *>(&desc.compress_mode), sizeof(desc.compress_mode));
    desc_io_.write(reinterpret_cast<const char *>(&desc.min_val), sizeof(desc.min_val));
    desc_io_.write(reinterpret_cast<const char *>(&desc.max_val), sizeof(desc.max_val));
    desc_io_.flush();
    file_size_ += SIZE_CHUNK_DESC;
    return chunk_id;
}

bool DataChunkDescIO::GetDescByChunkId(uint64_t chunk_id, ChunkDesc& desc) {
    uint64_t offset = chunk_id * SIZE_CHUNK_DESC;
    if (file_size_ <= offset) {
        LOG_DEBUG("chunk with id not exist %lu", chunk_id);
        LOG_DEBUG("%s file size: %ld", desc_file_.c_str(), FileUtil::GetFileSize(desc_file_));
        return false;
    }
    desc_io_.seekp(offset);
    desc_io_.read(reinterpret_cast<char *>(&desc.chunk_offset), sizeof(desc.chunk_offset));
    desc_io_.read(reinterpret_cast<char *>(&desc.chunk_size), sizeof(desc.chunk_size));
    desc_io_.read(reinterpret_cast<char *>(&desc.type), sizeof(desc.type));
    desc_io_.read(reinterpret_cast<char *>(&desc.compress_mode), sizeof(desc.compress_mode));
    desc_io_.read(reinterpret_cast<char *>(&desc.min_val), sizeof(desc.min_val));
    desc_io_.read(reinterpret_cast<char *>(&desc.max_val), sizeof(desc.max_val));
    return true;
}

int64_t DataChunkDescIO::GetChunkCount() {
    return file_size_/SIZE_CHUNK_DESC;
}

ChunkDesc DataChunkDescIO::GetDescByRecordId(uint64_t record_id, int64_t column_offset) {
    return ChunkDesc();
}

ChunkDesc DataChunkDescIO::GetDescByRecordId(uint64_t record_id) {
    return ChunkDesc();
}

}